package com.lq.dubbo.rocketmq.shop.mq;

import com.alibaba.fastjson.JSON;
import com.lq.dubbo.rocketmq.shop.constant.ShopCode;
import com.lq.dubbo.rocketmq.shop.entity.MQEntity;
import com.lq.dubbo.rocketmq.shop.mapper.TradeGoodsMapper;
import com.lq.dubbo.rocketmq.shop.mapper.TradeGoodsNumbersLogsMapper;
import com.lq.dubbo.rocketmq.shop.mapper.TradeMqConsumerLogMapper;
import com.lq.dubbo.rocketmq.shop.shop_pojo.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @program: dubborocketmqshop
 * @description: 消息监听
 * 因为订单回退,用户,商品,订单,优惠券系统登都需要监听做相应设置,所以messageModel(消费模式)设置为广播消费模式
 * @author: LQ
 * @create: 2021-06-24 16:25
 */
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}", consumerGroup = "${mq.order.consumer.group.name}", messageModel = MessageModel.BROADCASTING)
@Slf4j
public class CancelMqListener implements RocketMQListener<MessageExt> {

    @Autowired
    private TradeMqConsumerLogMapper mqConsumerLogMapper;

    @Autowired
    private TradeGoodsMapper goodsMapper;

    @Autowired
    TradeGoodsNumbersLogsMapper goodsNumbersLogsMapper;

    @Value("${mq.order.consumer.group.name}")
    private String groupName;

    // MessageExt 继承Message 有消息ID
    @Override
    public void onMessage(MessageExt messageExt) {
        String msgId = null;
        String tags = null;
        String keys = null;
        String body = null;
        try {
            // 1.解析消息的内容
            msgId = messageExt.getMsgId();
            tags = messageExt.getTags();
            keys = messageExt.getKeys();
            body = new String(messageExt.getBody(), "UTF-8");
            log.info("接收消息处理成功");
            // 2.查询消息消费记录
            TradeMqConsumerLogKey tradeMqConsumerLogKey = new TradeMqConsumerLogKey();
            tradeMqConsumerLogKey.setMsgKey(keys);
            tradeMqConsumerLogKey.setMsgTag(tags);
            tradeMqConsumerLogKey.setGroupName(groupName);
            TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(tradeMqConsumerLogKey);
            // 3.判断消息是否处理过
            if (mqConsumerLog != null) {
                // 3.1 处理过
                // 3.1.1 状态是成功或者正在处理(直接返回)
                Integer consumerStatus = mqConsumerLog.getConsumerStatus();
                if (ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode().intValue() == consumerStatus.intValue()) {
                    log.info("消息:" + msgId + ",已经处理过");
                    return;
                } else if (ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode().intValue() == consumerStatus.intValue()) {
                    log.info("消息:" + msgId + ",正在处理");
                    return;
                } else if (ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().intValue() == consumerStatus.intValue()) {
                    // 3.1.2 处理失败
                    Integer times = mqConsumerLog.getConsumerTimes();
                    // 3.1.2.1 处理次数大于三次,直接返回(操作重试机制限制次数)
                    if (times > 3) {
                        log.info("消息:" + msgId + ",消息处理超过3次,不能再进行处理了");
                        return;
                    }
                    // 3.1.2.2 小于三次
                    // 使用数据库乐观锁进行修改
                    mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                    TradeMqConsumerLogExample mqConsumerLogExample = new TradeMqConsumerLogExample();
                    TradeMqConsumerLogExample.Criteria criteria = mqConsumerLogExample.createCriteria();
                    criteria.andMsgTagEqualTo(mqConsumerLog.getMsgTag());
                    criteria.andMsgKeyEqualTo(mqConsumerLog.getMsgKey());
                    criteria.andGroupNameEqualTo(mqConsumerLog.getGroupName());
                    criteria.andConsumerTimesEqualTo(times);
                    int r = mqConsumerLogMapper.updateByExampleSelective(mqConsumerLog, mqConsumerLogExample);
                    if (r <= 0) {
                        // 未修改成功,有其他线程并发修改
                        log.info("并发修改异常,稍后重试!");
                        // return;
                    }
                }

            } else {
                // 3.2 没有处理
                // 3.2.1 设置消息内容
                mqConsumerLog = new TradeMqConsumerLog();
                mqConsumerLog.setMsgTag(tags);
                mqConsumerLog.setMsgKey(keys);
                mqConsumerLog.setGroupName(groupName);
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                mqConsumerLog.setMsgBody(body);
                mqConsumerLog.setMsgId(msgId);
                mqConsumerLog.setConsumerTimes(0);
                mqConsumerLogMapper.insert(mqConsumerLog);
            }
            // 4.回退库存
            MQEntity mqEntity = JSON.parseObject(body, MQEntity.class);
            Long goodsId = mqEntity.getGoodsId();
            TradeGoods goods = goodsMapper.selectByPrimaryKey(goodsId);
            goods.setGoodsNumber(goods.getGoodsNumber() + mqEntity.getGoodsNum());
            goodsMapper.updateByPrimaryKey(goods);
            // 5.记录库存操作日志
            TradeGoodsNumbersLogs goodsNumbersLogs = new TradeGoodsNumbersLogs();
            goodsNumbersLogs.setGoodsId(goodsId);
            goodsNumbersLogs.setOrderId(mqEntity.getOrderId());
            goodsNumbersLogs.setGoodsNumber(mqEntity.getGoodsNum());
            goodsNumbersLogs.setLogTime(new Date());
            goodsNumbersLogsMapper.insert(goodsNumbersLogs);
            // 6.记录消息消费日志(将消息处理状态改为成功)
            mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode());
            mqConsumerLog.setConsumerTimestamp(new Date());
            mqConsumerLogMapper.updateByPrimaryKey(mqConsumerLog);
            log.info("订单:" + mqEntity.getOrderId() + ":回退库存成功!");
        } catch (Exception e) {
            e.printStackTrace();
            TradeMqConsumerLogKey tradeMqConsumerLogKey = new TradeMqConsumerLogKey();
            tradeMqConsumerLogKey.setMsgKey(keys);
            tradeMqConsumerLogKey.setMsgTag(tags);
            tradeMqConsumerLogKey.setGroupName(groupName);
            TradeMqConsumerLog mqConsumerLog = mqConsumerLogMapper.selectByPrimaryKey(tradeMqConsumerLogKey);
            if (mqConsumerLog == null) {
                // 数据库未有记录(第一次失败)
                mqConsumerLog = new TradeMqConsumerLog();
                mqConsumerLog.setMsgTag(tags);
                mqConsumerLog.setMsgKey(keys);
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode());
                mqConsumerLog.setMsgBody(body);
                mqConsumerLog.setMsgId(msgId);
                mqConsumerLog.setConsumerTimes(1);
                mqConsumerLogMapper.insert(mqConsumerLog);
            } else {
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode());
                mqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes() + 1);
                mqConsumerLogMapper.updateByPrimaryKey(mqConsumerLog);
            }
        }
    }
}
